package drds.server.route.util;

import drds.server.util.StringUtil;

/**
 * 数据分区工具
 * 
 * @author 曹正辉<330937205@qq.com>
 */
public final class DataPartitionTool {

	/**
	 * 分区长度:数据段分布定义，其中取模的数一定要是2^n， 因为这里使用x % 2^n == x & (2^n - 1)等式，来优化性能。
	 */
	private static final int PARTITION_LENGTH = 1024;

	/**
	 * %转换为&操作的换算数值
	 */
	private static final long PARTITION_LENGTH_1 = PARTITION_LENGTH - 1;

	/**
	 * 分区线段
	 */
	private final int[] segments = new int[PARTITION_LENGTH];

	/**
	 * <pre>
	 * @param counts   数组中每个元素值表示定义的分区数。
	 * @param lengths  数组中每个元素值表示每个分区的取值长度。在这里counts数组的每个元素对应的分区数下的值相同。
	 * 注意：其中count,length两个数组的长度必须是一致的。
	 * 约束：1024 = sum((count[i]*length[i]))。
	 * </pre>
	 */
	public DataPartitionTool(int[] counts, int[] lengths) {
		if (counts == null || lengths == null || (counts.length != lengths.length)) {
			throw new IllegalArgumentException("error,check your scope & scopeLength definition.");
		}
		/**
		 * 分区数和
		 */
		int count = 0;
		int length = counts.length;
		for (int i = 0; i < length; i++) {
			count += counts[i];
		}
		int[] partitionArray = new int[count + 1];

		int index = 0;

		for (int i = 0; i < length; i++) {
			/**
			 * 相同分区数值的集合
			 */
			int $count = counts[i];
			for (int j = 0; j < $count; j++) {
				/**
				 * 该循环内数组的每个元素值相同。
				 */
				int $length = lengths[i];
				// ++index表示第一个有效值为1
				partitionArray[++index] = partitionArray[index - 1] + $length;
			}
		}
		if (partitionArray[partitionArray.length - 1] != PARTITION_LENGTH) {
			throw new RuntimeException("error,check your partitionScope definition.");
		}
		/**
		 * 所有的分区数个数
		 */
		int $length = partitionArray.length;
		for (int i = 1; i < $length; i++) {
			/**
			 * 将位于某个区域的[)的值建立第N分区的映射关系,依据是第N个分区有明确的开始和结束权重。
			 */
			for (int j = partitionArray[i - 1]; j < partitionArray[i]; j++) {
				segments[j] = (i - 1);
			}
		}
	}

	/**
	 * 通过hash路由到具体的节点
	 */
	public int hash(long hash) {
		return segments[(int) (hash & PARTITION_LENGTH_1)];
	}

	/**
	 * 通过hash路由到具体的节点
	 */
	public int hash(String key, int start, int end) {
		return hash(StringUtil.hash(key, start, end));
	}

}